package com.example.common.model.kafka;

/**
 * @author FanFu
 * @version 1.0
 * @date 2025/11/8 20:48
 */

import com.example.common.dto.User;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Slf4j
@Service
public class UserKafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private ObjectMapper objectMapper;

    @Value("${app.kafka.topics.user-register}")
    private String userRegisterTopic;

    @Value("${app.kafka.topics.user-update}")
    private String userUpdateTopic;

    /**
     * 发送用户注册消息
     */
    public void sendUserRegisterMessage(User user) {
        UserMessage message = UserMessage.createRegisterMessage(
                user.getId(), user.getUsername(), user.getEmail()
        );

        sendMessage(userRegisterTopic, String.valueOf(user.getId()), message);
        log.info("发送用户注册消息: {}", message);
    }

    /**
     * 发送用户更新消息
     */
    public void sendUserUpdateMessage(User user) {
        UserMessage message = UserMessage.createUpdateMessage(
                user.getId(), user.getUsername(), user.getEmail()
        );

        sendMessage(userUpdateTopic, String.valueOf(user.getId()), message);
        log.info("发送用户更新消息: {}", message);
    }

    /**
     * 发送消息通用方法
     */
    private void sendMessage(String topic, String key, Object message) {
        try {
            String messageJson = objectMapper.writeValueAsString(message);

            ListenableFuture<SendResult<String, String>> future =
                    kafkaTemplate.send(topic, key, messageJson);

            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onSuccess(SendResult<String, String> result) {
                    log.info("Kafka消息发送成功, topic: {}, partition: {}, offset: {}",
                            topic,
                            result.getRecordMetadata().partition(),
                            result.getRecordMetadata().offset());
                }

                @Override
                public void onFailure(Throwable ex) {
                    log.error("Kafka消息发送失败, topic: {}, message: {}", topic, message, ex);
                }
            });
        } catch (JsonProcessingException e) {
            log.error("Kafka消息JSON序列化失败: {}", message, e);
        }
    }

    /**
     * 发送用户删除消息
     */
    public void sendUserDeleteMessage(Long userId, String username) {
        UserMessage message = new UserMessage();
        message.setEventType("DELETE");
        message.setUserId(userId);
        message.setUsername(username);
        message.setEventTime(java.time.LocalDateTime.now());

        sendMessage(userRegisterTopic, String.valueOf(userId), message);
        log.info("发送用户删除消息: {}", message);
    }
}